package com.showdor.springboot.jingqing;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.security.scram.ScramLoginModule;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;

/**
 * @author showdor
 * @email chuanqi@outlook.com
 * @date 2025/3/10
 */


public class KafkaDataConsumer {

    // 配置参数（与消费者配置对应）
    private static final String BOOTSTRAP_SERVERS = "115.238.84.147:9093";
    private static final String TOPIC = "external.wenZhou.eventRescue";
    private static final String USERNAME = "guangxin.quzhou";
    private static final String PASSWORD = "78K%!4g8LV";

    public static void main(String[] args) {
        // 1. 生产者配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test_gx");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // SASL认证配置
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.mechanism", "SCRAM-SHA-256");
        props.put("sasl.jaas.config", ScramLoginModule.class.getName() + " required username=\"" + USERNAME + "\" password=\"" + PASSWORD + "\";");

        // 2. 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        ;
        // 3. 输出消息获得的消息日志
        consumer.subscribe(Collections.singleton(TOPIC));
        while (true) {
            consumer.poll(1000);
            for (ConsumerRecord<String, String> record : consumer.poll(1000)) {
                System.out.println("Received message: " + record.value());
            }
        }

    }
}